package com.zhixing.order.listener;

import com.zhixing.order.domain.entity.OrderSendInfo;
import com.zhixing.order.domain.enums.OrderStatus;
import com.zhixing.order.domain.vo.OrderGrabInfoVO;
import com.zhixing.order.domain.vo.OrderStatusVO;
import com.zhixing.order.mapper.OrderMapper;
import com.zhixing.order.service.OrderService;
import com.zhixing.order.websocket.DriverWebSocket;
import com.zhixing.order.websocket.OrderWebSocket;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.retrytopic.RetryTopicHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.retry.RetryException;
import org.springframework.retry.annotation.Backoff;
import org.springframework.stereotype.Component;
import static com.zhixing.common.constant.Constant.*;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderListener {

    private final OrderMapper orderMapper;
    private final OrderService orderService;
    private final DriverWebSocket driverWebSocket;
    private final OrderWebSocket orderWebSocket;
    private final RedisTemplate<String, Object> redisTemplate;

    @RetryableTopic(attempts = DISPATCH_COUNT_ADD_ONE, include = RetryException.class,
            backoff = @Backoff(delay = OVERTIME_MILLISECONDS / DISPATCH_COUNT))
    @KafkaListener(topics = ZHIXING_ORDER_SUBMIT, groupId = ZHIXING_ORDER_DISPATCH)
    public void dispatchOrder(Long orderId, @Header(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS) Integer attempt) {
        log.info("订单服务监听，向附近司机派单，订单id：{}", orderId);
        OrderSendInfo orderSendInfo = orderMapper.getOrderSendInfo(orderId);
        if (orderSendInfo.getOrderStatus() == OrderStatus.WAIT) {
            if (attempt < 6) {
                orderService.dispatchOrder(orderSendInfo);
                throw new RetryException("继续派单");
            }
            else orderService.cancelOrder(orderId);
        }
    }

    @KafkaListener(topics = ZHIXING_SEND_DRIVER + "${spring.application.instance-id}", groupId = ZHIXING_DRIVER_RECEIVE + "${spring.application.instance-id}")
    public void driverReceiveOrder(OrderGrabInfoVO orderGrabInfoVO) {
        log.info("订单服务监听，接收新订单并推送给司机，司机id：{}", orderGrabInfoVO.getDriverId());
        driverWebSocket.sendMessage(orderGrabInfoVO);
    }

    @KafkaListener(topics = ZHIXING_ORDER_DELETE_FAIL, groupId = ZHIXING_ORDER_DELAY_DELETE)
    public void driverReceiveOrder(Long orderId) {
        log.info("订单服务监听，订单延迟删除，订单id：{}", orderId);
        redisTemplate.delete(ZHIXING_SUBMIT_ORDER + orderId);
    }

    @KafkaListener(topics = ZHIXING_SEND_PASSENGER + "${spring.application.instance-id}", groupId = ZHIXING_PASSENGER_RECEIVE + "${spring.application.instance-id}")
    public void passengerReceiveOrder(OrderStatusVO orderStatusVO) {
        log.info("订单服务监听，接收订单变化并推送给乘客，订单id：{}", orderStatusVO.getOrderId());
        orderWebSocket.sendMessage(orderStatusVO);
    }

    @KafkaListener(topics = ZHIXING_ORDER_COMPLETE, groupId = ZHIXING_DRIVER_REMIT)
    public void driverRemit(Long orderId) {
        log.info("订单服务监听，司机打款，订单id：{}", orderId);
        orderService.driverRemit(orderId);
    }
}
